package cn.lzd.mr.case9;

import com.alibaba.fastjson.JSONObject;
import org.apache.commons.beanutils.BeanUtils;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.LocatedFileStatus;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.fs.RemoteIterator;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.mapreduce.lib.output.MultipleOutputs;
import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;

import java.io.IOException;
import java.lang.reflect.InvocationTargetException;
import java.util.*;

/**
 * Created by lzd on 2019/3/25.
 * ids合并处理
 */
public class IdsMR {

    static class Mp extends Mapper<LongWritable, Text, Text, IdsBean> {
        Text outKey = new Text();

        @Override
        protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
            String val = value.toString();
            IdsBean idsBean = JSONObject.parseObject(val, IdsBean.class);

            outKey.set(idsBean.getCompanyId());
            context.write(outKey, idsBean);
        }

    }

    static class Rd extends Reducer<Text, IdsBean, Text, NullWritable> {
        Text outkey = new Text();

        // 输出类型和Reduce一致
        private MultipleOutputs<Text, NullWritable> mos;

        @Override
        protected void setup(Context context) throws IOException, InterruptedException {
            mos = new MultipleOutputs<>(context);
        }

        @Override
        protected void reduce(Text key, Iterable<IdsBean> values, Context context) throws IOException, InterruptedException {
            IdsBean outIdsBean = new IdsBean();
            Iterator<IdsBean> iterator = values.iterator();
            List<String> list = new ArrayList<>();
            boolean tag = true;
            while (iterator.hasNext()) {
                IdsBean idsBean = iterator.next();
                if (tag) {
                    try {
                        BeanUtils.copyProperties(outIdsBean, idsBean);
                        tag = false;
                    } catch (IllegalAccessException e) {
                        e.printStackTrace();
                    } catch (InvocationTargetException e) {
                        e.printStackTrace();
                    }
                }
                list.addAll(idsBean.ids2List());
            }
            Map<String, Set<String>> setMap = null;
            JSONObject jsonStr = new JSONObject();
            jsonStr.put("testjsonStr", "hello jsonStr");
            JSONObject jsonCount = new JSONObject();
            jsonCount.put("testjsonCount","hello jsoncount");
//            for (String mKey : setMap.keySet()) {
//                jsonStr.put(mKey, setMap.get(mKey).toArray());
//                jsonCount.put(mKey, setMap.get(mKey).size());
//            }


            /**
             * 重点来了，输出文件带标识符
             */
            outIdsBean.setIds(jsonStr.toJSONString());
            outkey.set(JSONObject.toJSONString(outIdsBean));
            mos.write("idsAll", outkey, NullWritable.get(), "F:\\ids\\out");

            outIdsBean.setIds(jsonCount.toJSONString());
            outkey.set(JSONObject.toJSONString(outIdsBean));
            mos.write("idsAllCount", outkey, NullWritable.get(), "F:\\idsCount\\out");
        }

        @Override
        protected void cleanup(Context context) throws IOException, InterruptedException {
            mos.close();
        }

        private void sendKafka(IdsBean idsBean, JSONObject json) {
            try {
                IdsBean out = new IdsBean();
                BeanUtils.copyProperties(out, idsBean);
                out.setIds(json.toJSONString());

                //TODO kafka
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
    }

    public static void main(String[] args) throws IOException, ClassNotFoundException, InterruptedException {
        Configuration configuration = new Configuration();
        Job job = Job.getInstance(configuration);

        job.setJarByClass(IdsMR.class);

        job.setMapperClass(Mp.class);
        job.setMapOutputKeyClass(Text.class);
        job.setMapOutputValueClass(IdsBean.class);

        job.setReducerClass(Rd.class);
        job.setOutputKeyClass(Text.class);
        job.setOutputValueClass(NullWritable.class);


        MultipleOutputs.addNamedOutput(job, "idsAll", TextOutputFormat.class, Text.class, NullWritable.class);
        MultipleOutputs.addNamedOutput(job, "idsAllCount", TextOutputFormat.class, Text.class, NullWritable.class);


//        FileInputFormat.setInputPaths(job, new Path(Dict.TablesPath.HDFS_FILE_PATH_WHOLE_IDS_IN));
//        FileOutputFormat.setOutputPath(job, new Path(Dict.TablesPath.HDFS_FILE_PATH_WHOLE_IDS_OUT));


        List<Path> paths = filterFile(configuration);
        FileInputFormat.setInputPaths(job, paths.toArray(new Path[paths.size()]));
        FileOutputFormat.setOutputPath(job, new Path("F:\\ids"));

        System.exit(job.waitForCompletion(true) ? 0 : 1);
    }


    private static List<Path> filterFile(Configuration configuration) throws IOException {
        String matchRule = "Ids-";
        String test_path = "G:\\ids";

        Path path = new Path(test_path);
        FileSystem fileSystem = path.getFileSystem(configuration);
        RemoteIterator<LocatedFileStatus> locatedFileStatusRemoteIterator = fileSystem.listFiles(path, true);

        List<Path> list = new ArrayList<>();

        while (locatedFileStatusRemoteIterator.hasNext()) {
            LocatedFileStatus file = locatedFileStatusRemoteIterator.next();
            String fileName = file.getPath().getName();
            if (fileName.indexOf(matchRule) != -1 && file.getLen() > 0) {
                list.add(file.getPath());
            }
        }
        return list;
    }
}